#ifndef MQSERVER_MQ_CHANNEL_HPP_
#define MQSERVER_MQ_CHANNEL_HPP_
#include "mq_consumer.hpp"
#include "mq_exchange.hpp"
#include "mq_host.hpp"
#include "mq_route.hpp"
#include "../mqcomm/mq_threadpool.hpp"
#include "../mqcomm/proto/mq_protocol.pb.h"
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/proto/codec.h"
#include <functional>
#include <iostream>
#include <memory>
#include <mutex>
#include <string>
#include <unordered_map>
#include <utility>

namespace mqserver 
{
class Channel;
class ChannelManager;
using ChannelPtr = std::shared_ptr<Channel>;
using ChannelManagerPtr = std::shared_ptr<ChannelManager>;
using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;

using openChannelRequestPtr = std::shared_ptr<proto::protocol::openChannelRequest>;
using declareExchangeRequestPtr = std::shared_ptr<proto::protocol::declareExchangeRequest>;
using closeChannelRequestPtr = std::shared_ptr<proto::protocol::closeChannelRequest>;
using deleteExchangeRequestPtr = std::shared_ptr<proto::protocol::deleteExchangeRequest>;
using declareQueueRequestPtr = std::shared_ptr<proto::protocol::declareQueueRequest>;
using deleteQueueRequestPtr = std::shared_ptr<proto::protocol::deleteQueueRequest>;
using queueBindRequestPtr = std::shared_ptr<proto::protocol::queueBindRequest>;
using queueUnBindRequestPtr = std::shared_ptr<proto::protocol::queueUnBindRequest>;
using basicPublishRequestPtr = std::shared_ptr<proto::protocol::basicPublishRequest>;
using basicAckRequestPtr = std::shared_ptr<proto::protocol::basicAckRequest>;
using basicConsumeRequestPtr = std::shared_ptr<proto::protocol::basicConsumeRequest>;
using basicCancelRequestPtr = std::shared_ptr<proto::protocol::basicCancelRequest>;


class Channel
{
public:
	Channel(const std::string& cid,
		    const ProtobufCodecPtr& codec,
		    const muduo::net::TcpConnectionPtr& conn,
		    const VirtualHostPtr& host,
		    const ConsumerManagerPtr& cmp,
		    const pool::ThreadPoolPtr& pool)
		: channel_id_(cid), conn_(conn), codec_(codec), consumer_mg(cmp), host_(host), pool_(pool)
	{
        LOG_DEBUG << "new channel: " << this << '\n';
    }
    ~Channel()  // 信道关闭, 删除消息的订阅者
    {
        if(consumer_.get()) {
			consumer_mg->remove(consumer_->qname_, consumer_->tag_);
		}
        LOG_DEBUG << "del channel: " << this << '\n';
    }

    // 涉及到数据操作, 都使用请求对象进行传参(序列化), 而不是传递大量构造对象参数
    // 声明&删除交换机
    void declareExchange(const declareExchangeRequestPtr& req)
    {
        bool ret = host_->declareExchange(req->exchange_name(), req->exchange_type(), \
                            req->durable(), req->auto_delete(), req->args());
        // 针对请求id和信道id进行响应  
        basicResponse(ret, req->rid(), req->channel_id());    
    }
    void deleteExchange(const deleteExchangeRequestPtr& req)
    {
        host_->deleteExchange(req->exchange_name());
        basicResponse(true, req->rid(), req->channel_id());    
    }
    // 声明&删除队列
    void declareQueue(const declareQueueRequestPtr& req)
    {
        bool ret = host_->declareQueue(req->queue_name(), req->durable(), 
                            req->exclusive(), req->auto_delete(), req->args());

        // 队列创建成功, 初始化队列消费者管理句柄
		if (ret) consumer_mg->initQueueConsumer(req->queue_name());
		basicResponse(ret, req->rid(), req->channel_id());    
    }
    void deleteQueue(const deleteQueueRequestPtr& req)
    {
        host_->deleteQueue(req->queue_name());
        // 队列没了, 订阅该队列的消费者也就没有存在的意义
		consumer_mg->destoryQueueConsumer(req->queue_name());
		basicResponse(true, req->rid(), req->channel_id());    
    }
    // 绑定&解绑队列
    void QueueBind(const queueBindRequestPtr& req)
    {
        bool ret = host_->bind(req->exchange_name(), req->queue_name(), req->binding_key());
        basicResponse(ret, req->rid(), req->channel_id());    
    }
    void QueueUnbind(const queueUnBindRequestPtr& req)
    {
        host_->unbind(req->exchange_name(), req->queue_name());
        basicResponse(true, req->rid(), req->channel_id()); 
    }
    // 发布&确认消息
    void basicPublish(const basicPublishRequestPtr& req)
    {
        // 1. 获取交换机信息
        ExchangePtr exchange = host_->selectExchange(req->exchange_name());
        if(exchange.get() == nullptr) {
            basicResponse(false, req->rid(), req->channel_id());
        }
        // 2. 获取交换机绑定信息, 根据交换路由规则, 将消息路由到绑定的队列
        proto::message::BasicProperties* properties = nullptr;
        std::string routing_key;
        if(req->has_properties()) 
        { 
            properties = req->mutable_properties();
            routing_key = properties->routing_key();
        }
        MsgQueueBindingMap msgque_binding_map = host_->getExchangeBindings(exchange->name_);
        for(auto& [qname, binding] : msgque_binding_map)
        {
            // std::cout << "qname: " << qname << '\n';
            // std::cout << "qname: " << binding->msgqueue_name_ << '\n';
            if(Router::route(exchange->type_, routing_key, binding->binding_key_)) 
            {
                // 3. 将消息加入添加到队列
                host_->basicPublish(qname, properties, req->body());
                // 4. 向线程池提交一个消费任务(向指定队列推送消息)
                pool_->submitTask(std::bind(&Channel::consume, this, qname));
            }
        }
        basicResponse(true, req->rid(), req->channel_id());
    }
    void basicAck(const basicAckRequestPtr& req)
    {
        // 对指定队列的指定消息进行确认
        host_->basicAck(req->queue_name(), req->message_id());
        basicResponse(true, req->rid(), req->channel_id());
    }
    // 订阅&取消订阅队列消息
    void basicConsume(const basicConsumeRequestPtr& req)
    {
        // 1. 判断队列是否存在
        if(host_->existsQueue(req->queue_name()) == false) {
            basicResponse(false, req->rid(), req->channel_id());
        }
        // 2. 新增队列消费者
        auto cb = std::bind(&Channel::callback, this, 
                        std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
		consumer_ = consumer_mg->create(req->consumer_tag(), req->queue_name(), req->auto_ack(), cb);
		basicResponse(true, req->rid(), req->channel_id());
    }
    void basicCancel(const basicCancelRequestPtr& req)
    {
        // 取消指定队列中的订阅者
		consumer_mg->remove(req->queue_name(), req->consumer_tag());
		basicResponse(true, req->rid(), req->channel_id());
    }
private:
    // 消息推送请求, 将消息推送给channel对应的客户端
    void callback(const std::string& tag,  
                  const proto::message::BasicProperties* properties,    
                  const std::string& body)
    {
        proto::protocol::basicConsumeResponse resp;
        // 设置响应请求参数
        resp.set_channel_id(channel_id_);
        resp.set_consumer_tag(tag);
        resp.set_body(body);
        if(properties) {
            resp.mutable_properties()->set_id(properties->id());
            resp.mutable_properties()->set_delivery_mode(properties->delivery_mode());
            resp.mutable_properties()->set_routing_key(properties->routing_key());
        }
        // 通过协议处理器发送 
        codec_->send(conn_, resp);
    }
    void consume(const std::string& qname)  // 线程执行消费的函数
    {
        // 1. 从队列获取消息(消费一条消息)
        MessagePtr message = host_->basicConsume(qname);
        if(!message.get()) {
            LOG_ERROR << "consume task fail, there is no message in the \"" << qname << "\"\n";
            return;
        }
        // 2. 从队列订阅者中获取订阅者
		ConsumerPtr consumer = consumer_mg->getConsumer(qname);
		if(!consumer.get()) {
            LOG_ERROR << "consume task fail, there is no consumer in the \"" << qname << "\"\n";
            return;
        }
        // 3. 回调订阅者的消息处理函数, 推送消息
        consumer->callback_(consumer->tag_, message->mutable_payload()->mutable_properties(), message->payload().body());
        // 4. 消息是否自动确认
        if(consumer->auto_ack_) host_->basicAck(qname, message->payload().properties().id());
    }

    void basicResponse(bool ok, const std::string& rid, const std::string& channel_id)
    {
        proto::protocol::basicCommonResponse resp;
        resp.set_rid(rid);
        resp.set_channel_id(channel_id);
        resp.set_ok(ok);
        codec_->send(conn_, resp);
    }
private:
    std::string channel_id_;                // 信道标识id
    ConsumerPtr consumer_;                  // 信道关联的消费者
    muduo::net::TcpConnectionPtr conn_;     // 通信连接
    ProtobufCodecPtr codec_;                // protobuf协议处理句柄
	ConsumerManagerPtr consumer_mg;         // 消费者管理句柄
	VirtualHostPtr host_;                   // 虚拟机句柄
    pool::ThreadPoolPtr pool_;                
};
    
class ChannelManager
{
public:
    ChannelManager() = default;
    bool openChannel(const std::string& cid, 
                    const ProtobufCodecPtr& codec,
                    const muduo::net::TcpConnectionPtr& conn,
                    const VirtualHostPtr& host, 
                    const ConsumerManagerPtr& cmp,
                    const pool::ThreadPoolPtr& pool)
    {
        std::lock_guard<std::mutex> lock(mtx_);
        if(existsChannel(cid)) return false;
        auto channel = std::make_shared<Channel>(cid, codec, conn, host, cmp, pool);
        channels_.emplace(cid, channel);
        return true;
    }
    void closeChannel(const std::string& cid)
    {
        std::lock_guard<std::mutex> lock(mtx_);
        channels_.erase(cid);
    }
    ChannelPtr getChannel(const std::string& cid)
    {
        std::lock_guard<std::mutex> lock(mtx_);
        if(!existsChannel(cid)) return ChannelPtr();
        return channels_[cid];
    }
private:
    bool existsChannel(const std::string& cid)
    {
        return channels_.find(cid) != channels_.end();
    }
private:
    std::mutex mtx_;
    std::unordered_map<std::string, ChannelPtr> channels_;  // <channel_id, channel>
};

}

#endif